package com.qyer.search.engine;

import static com.qyer.commons.utils.CommonUtils.LEFT;
import static com.qyer.commons.utils.CommonUtils.RIGHT;
import static com.qyer.commons.utils.CommonUtils.splitList;

import com.qyer.log.WarningLogger;
import com.qyer.search.cache.RedisCacheHolder;
import com.qyer.search.enums.FieldsMode;
import com.qyer.search.model.SearchParameter;
import com.qyer.search.searchplan.SQPlan2;
import com.qyer.words.model.Word;
import org.apache.commons.collections4.CollectionUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.List;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ThreadLocalRandom;

/**
 * Created by tangzheng on 16/6/15.
 */

public class RecursiveSearchASSetTask extends RecursiveTask<Set<String>> {

  private static final int TIMEOUT_IN_SECONDS = 60;
  private static final int TIMEOUT_IN_SECONDS_3 = 20;

  private AbstractESEngine searchEngine;

  private List<SQPlan2> plans;

  private boolean showHighlight;
  private FieldsMode mode;

  /*protected String getCacheKey(SearchParameter sp, final List<Word> words) {
    int result = sp.getConditionCacheKey();
    if (CollectionUtils.isEmpty(words)) {
      return String.valueOf(result);
    }
    List<Word> newList = new ArrayList<>(words);
    Collections.sort(newList);
    result = 31 * result + newList.hashCode();
    //basic 是否区分 fix
    return "sr." + (simpleResult ? "0" : "1") + String.valueOf(result);
  }*/

  protected String getCacheKeyLong(SearchParameter sp, final List<Word> words) {
    long result = sp.getZCacheKey();
    if (CollectionUtils.isEmpty(words)) {
      return "v3.test.null."+ mode + String.valueOf(result);
    }
    long result2 = 0;
    List<Word> newList = new ArrayList<>(words);
    Collections.sort(newList);
    result = 31 * result + newList.hashCode();
    //basic 是否区分 fix
    return "v3.test.so." + mode + String.valueOf(result);
  }

  public RecursiveSearchASSetTask(AbstractESEngine searchEngine, List<SQPlan2> plans,
                                  boolean showHighlight, FieldsMode mode) {
    if (CollectionUtils.isEmpty(plans)) {
      throw new RuntimeException("Empty plan.");
    }
    this.searchEngine = searchEngine;
    this.plans = plans;
    this.showHighlight = showHighlight;
    this.mode = mode;
  }

  public Set<String> atomicPlanSearch(SQPlan2 plan) {

    //System.out.println("atomicPlanSearch go~");
    long t1 = System.currentTimeMillis();
    SearchParameter sp = plan.getSearchParameter();
    String key = getCacheKeyLong(sp, plan.getWords());
    System.out.println("key:"+key);

    long t8 = System.currentTimeMillis();
    List<String> result_redis = RedisCacheHolder.getInstance().getCache(key);
    long t9 = System.currentTimeMillis();
    //System.out.println("redis get Time:"+(t9 - t8));
    Set<String> result = new HashSet<>();
    try {
      if (CollectionUtils.isEmpty(result_redis)) {
        WarningLogger.logWarning(key + ":cache_missed", sp, null);
        System.out.println(key + ":cache_missed");
        long t3 = System.currentTimeMillis();
        result = searchEngine.executePlanExtractAsSet(sp, plan, showHighlight, mode);
        long t4 = System.currentTimeMillis();
        //System.out.println("real search time:"+(t4 - t3));
        if (CollectionUtils.isNotEmpty(result)) {
        }
        result_redis = new ArrayList<>();
        if (result != null) {
          result_redis.addAll(result);
          int lize = 0;
          for(String str:result_redis){
            lize = lize + str.getBytes().length;
          }
          System.out.println("result bytes length:"+lize);
        }
        //RedisCacheHolder.getInstance().putCache(key, result_redis, TIMEOUT_IN_SECONDS);
        long t5 = System.currentTimeMillis();
        RedisCacheHolder.getInstance().putCache(key, result_redis,
                                                (TIMEOUT_IN_SECONDS + (ThreadLocalRandom.current().nextInt(0, TIMEOUT_IN_SECONDS_3))));
        long t6 = System.currentTimeMillis();
        //System.out.println("redis store time:"+(t6 - t5));
      } else {
        WarningLogger.logWarning(key + ":cache_hit", sp, null);
        System.out.println(key + ":cache_hit");
        result.addAll(result_redis);
        AbstractESEngine.logDebugMap(sp, true, plan);
      }
    }catch (Exception e){
      e.printStackTrace();
      System.out.println(e.toString());
    }
    /*if (result == null) {
      result = Collections.emptyList();
    }*/
    long t2 = System.currentTimeMillis();
    //System.out.println("atomic search time:"+(t2-t1));
    return result;
  }

  @Override protected Set<String> compute() {
    Set<String> result;
    if (CollectionUtils.isEmpty(plans)) {
      result = Collections.emptySet();
    } else if (plans.size() == 1) {
      result = atomicPlanSearch(plans.get(0));
      if(CollectionUtils.isEmpty(result)) {
        result = new HashSet<>();
        //System.out.println("ES null");
      }else {
        //System.out.println("ES ONLY TASK result size:"+result.size());
      }
    } else {
      List<SQPlan2> l = splitList(plans, LEFT);
      RecursiveSearchASSetTask left = new RecursiveSearchASSetTask(searchEngine, l, showHighlight, mode);
      l = splitList(plans, RIGHT);
      RecursiveSearchASSetTask right = new RecursiveSearchASSetTask(searchEngine, l, showHighlight, mode);
      left.fork();
      right.fork();
      int mergeSize = 0;
      Set<String> leftResult = left.join(), rightResult = right.join();
      mergeSize += CollectionUtils.isEmpty(leftResult) ? 0 : leftResult.size();
      mergeSize += CollectionUtils.isEmpty(rightResult) ? 0 : rightResult.size();
      result = new HashSet<>(mergeSize);
      if(CollectionUtils.isNotEmpty(leftResult)) {
        result.addAll(leftResult);
      }
      if(CollectionUtils.isNotEmpty(rightResult)) {
        result.addAll(rightResult);
      }
      //System.out.println("TASK result size:"+result.size());
    }
    return result;
  }

}

